package com.mygame.message.nats;
import com.mygame.common.cloud.PlayerServiceInstance;
import com.mygame.common.codec.DataCodecKit;
import com.mygame.common.concurrent.GameEventExecutorGroup;
import com.mygame.message.dispatcher.GameMessageService;
import com.mygame.common.msg.MessageType;
import com.mygame.message.channel.GameChannelConfig;
import com.mygame.message.channel.GameChannelInitializer;
import com.mygame.message.channel.GameMessageEventDispatchService;
import com.mygame.message.channel.IMessageSendFactory;
import com.mygame.common.protobuf.ExternalMessage;
import com.mygame.message.rpc.GameRpcService;
import io.nats.streaming.Message;
import io.nats.streaming.StreamingConnection;
import io.netty.util.concurrent.DefaultEventExecutorGroup;
import io.netty.util.concurrent.EventExecutorGroup;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Service;
import java.io.IOException;

import com.mygame.common.natsConfig.Subscribe;

/**
 * 用于接收网关消息，并分发消息到业务中。
 */
@Service
public class NatsGatewayMessageConsumerService {
    private IMessageSendFactory gameGatewayMessageSendFactory;// 默认实现的消息发送接口，GameChannel返回的消息通过此接口发送到kafka中
    private GameRpcService gameRpcSendFactory;
    private Logger logger = LoggerFactory.getLogger(NatsGatewayMessageConsumerService.class);
    @Autowired
    private GameChannelConfig serverConfig;// GameChannel的一些配置信息
    @Autowired
    private GameMessageService gameMessageService; // 消息管理类，负责管理根据消息id，获取对应的消息类实例
    private GameMessageEventDispatchService gameChannelService;// 消息事件分类发，负责将用户的消息发到相应的GameChannel之中。
    private GameEventExecutorGroup workerGroup;// 业务处理的线程池
    private EventExecutorGroup rpcWorkerGroup = new DefaultEventExecutorGroup(2);
    @Autowired
    private PlayerServiceInstance playerServiceInstance;
    @Autowired
    private ApplicationContext context;
    @Autowired
    private StreamingConnection connection;


    public void start(GameChannelInitializer gameChannelInitializer, int localServerId) {
        workerGroup = new GameEventExecutorGroup(serverConfig.getWorkerThreads());

        try {
            gameGatewayMessageSendFactory = new NatsGatewayMessageSendFactory(connection, serverConfig.getGatewayGameMessageTopic());
            gameRpcSendFactory = new GameRpcService(serverConfig.getRpcRequestGameMessageTopic(),serverConfig.getRpcResponseGameMessageTopic(),localServerId, playerServiceInstance, rpcWorkerGroup, connection);
        }catch (Exception e){

        }
        gameChannelService = new GameMessageEventDispatchService(context,workerGroup, gameGatewayMessageSendFactory, gameRpcSendFactory, gameChannelInitializer);
        workerGroup = new GameEventExecutorGroup(serverConfig.getWorkerThreads());
    }

    //business-game-message-topic-10101
    @Subscribe(value = "business-game-message-topic", durableName = "bar", queue = "queue")
    public void consumeGameRequest(Message message) throws IOException {
        logger.debug("收到游戏服消息,sub={}", message.getSubject());
        ExternalMessage gameMessage = this.getGameMessage(MessageType.REQUEST, message.getData());
        gameChannelService.fireReadMessage(gameMessage.getPlayerId(), gameMessage);
    }
    @Subscribe(value = "rpc-request-game-message-topic", durableName = "bar", queue = "queue")
    public void consumeRPCRequest(Message message) throws IOException {
        logger.debug("收到rpc-request消息,sub={}", message.getSubject());
        ExternalMessage gameMessage = this.getGameMessage(MessageType.RPC_REQUEST, message.getData());
        gameChannelService.fireReadRPCRequest(gameMessage);
    }

    // sendRpcResponse 方法会返回rpc响应
    @Subscribe(value = "rpc-response-game-message-topic", durableName = "bar", queue = "queue")
    public void consumeRPCResponse(Message message) throws IOException {
        logger.debug("收到rpc-response消息,sub={}", message.getSubject());
        ExternalMessage gameMessage = this.getGameMessage(MessageType.RPC_RESPONSE, message.getData());
        this.gameRpcSendFactory.receiveRpcResponse(gameMessage);
    }

    private ExternalMessage getGameMessage(MessageType mesasageType, byte[] data) {
        ExternalMessage externalMessage = DataCodecKit.decode(data, ExternalMessage.class);
        logger.debug("收到{}消息：{}", mesasageType, externalMessage.getMessageId());
        return externalMessage;
    }



}
